package com.doit.demo.day02.sources;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.util.Random;

/**
 * @DATE 2022/2/15/16:53
 * @Author MDK
 * @Version 2021.2.2
 *
 *  自定义并行的Source
 *  *  第一种,实现ParallelSourceFunction接口
 *  *  第二种,继承RichParallelSourceFunction抽象类
 *
 *  如果继承RichParallelSourceFunction抽象类,重写open和close方法,方法的调用顺序
 *  open(执行一次) -> run(执行一次,但可以在该方法中用while循环产生数据) -> cancel(执行一次) ->close(执行一次)
 *
 *  ????为什么ParallelSourceFunction的并行度是8,而RichParallelSourceFunction并行度却是1??????
 **/
public class CustomParSource2 {
    public static void main(String[] args) throws Exception {
        Configuration configuration = new Configuration();
        configuration.setInteger("rest.port",8081);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<String> nums = env.addSource(new MyParSource());
        System.out.println("继承RichParallelSourceFunction自定义Source的并行度:"+ nums.getParallelism());

        nums.print();
        env.execute();
    }

    private static class MyParSource extends RichParallelSourceFunction<String>{

        boolean flag = true;

        @Override
        public void open(Configuration parameters) throws Exception {
            System.out.println("open method invoked ~~~~~~");
        }

        @Override
        public void run(SourceContext<String> sourceContext) throws Exception {
            System.out.println("run method invoked !!!!!");
            Random random = new Random();
            while (flag){
                Thread.sleep(2000);
                sourceContext.collect("subtask:"+ getRuntimeContext().getIndexOfThisSubtask()+ " -> " + random.nextInt() + "");
            }
        }

        @Override
        public void cancel() {
            //web UI界面或者命令行执行cancel命令
            System.out.println("cancel method invoked ######");
            flag = false;
        }

        @Override
        public void close() throws Exception {
            System.out.println("close method invoked $$$$$");
        }

    }
}
